package com.cetccloud.ap.proxy.channel.cossDomain;

import cn.hutool.core.bean.BeanUtil;
import com.cetccloud.ap.proxy.channel.Channel;
import com.cetccloud.ap.proxy.channel.cossDomain.model.CallTypeEnum;
import com.cetccloud.ap.proxy.channel.cossDomain.model.CrossDomainModel;
import com.cetccloud.ap.proxy.common.ProxyContext;
import com.cetccloud.ap.proxy.coord.Coord;
import com.cetccloud.ap.proxy.processor.Processor;
import com.cetccloud.ap.proxy.processor.http.model.HttpCallParam;
import com.cetccloud.ap.proxy.processor.kafka.model.KafkaTransferParam;
import com.cetccloud.ap.proxy.processor.model.ProcessResult;
import com.cetccloud.ap.proxy.processor.subscribe.model.SubscribeHttpParam;
import com.cetccloud.ap.util.exception.ApException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

/**
 * @ClassName CrossDomainDispatcher
 * @Description CrossDomain消费Dispatcher
 * @Author superz
 * @Date 2021/4/9 15:27
 * @Version 1.0
 **/
@Slf4j
@Component
public class CrossDomainDispatcher {

    public static final String SRC_ZONE_ID = "SRC_ZONE_ID";

    @Autowired
    private Channel crossDomainChannel;

    @Autowired
    @Qualifier("srcCoord")
    private Coord srcCoord;

    @Autowired
    @Qualifier("httpCallProcessor")
    private Processor httpCallProcessor;

    @Autowired
    @Qualifier("subscribeHttpProcessor")
    private Processor subscribeHttpProcessor;

    @Autowired(required = false)
    @Qualifier("kafkaTransferProcessor")
    private Processor kafkaTransferProcessor;

    public Mono<?> dispatch(CrossDomainModel model) {
        log.info("### CrossDomainDispatcher dispatch ### {}, [{}], {}, {}", model.getTransId(), model.getCallType(),
                model.getSrcAppId() + "->" + model.getSrcAppId(), model.getDestAppId() + "->" + model.getDestAppId());

        ProxyContext context = new ProxyContext();
        context.setAttribute(SRC_ZONE_ID, model.getSrcZoneId());

        return Mono.create(_sink -> {
            try {
                switch (model.getCallType()) {
                    case HTTP_REQ:
                        HttpCallParam httpCallParam = BeanUtil.mapToBean(model.getPayload(), HttpCallParam.class, false, null);
                        Mono<ProcessResult> httpCallMono = httpCallProcessor.process(httpCallParam, context);
                        httpResp(model, httpCallMono);
                        break;
                    case HTTP_REQ_SUBSCRIBE:
                        SubscribeHttpParam subscribeHttpParam = BeanUtil.mapToBean(model.getPayload(), SubscribeHttpParam.class, false, null);
                        Mono<ProcessResult> subscribeCallMono = subscribeHttpProcessor.process(subscribeHttpParam, context);
                        httpResp(model, subscribeCallMono);
                        break;
                    // 通用的http响应处理
                    case HTTP_RESP:
                        String transId = model.getTransId();
                        ProcessResult processResult = BeanUtil.mapToBean(model.getPayload(), ProcessResult.class, false, null);
                        if (processResult.isSuccess()) {
                            srcCoord.success(transId, processResult.getBody());
                        } else {
                            srcCoord.error(transId, processResult.getMsg());
                        }
                        break;
                    case KAFKA_TRANSFER:
                        if (null == kafkaTransferProcessor) {
                            log.warn("kafka profile未启用");
                            break;
                        }
                        KafkaTransferParam kafkaTransferParam = BeanUtil.mapToBean(model.getPayload(), KafkaTransferParam.class, false, null);
                        Mono<Void> kafkaTransferMono = kafkaTransferProcessor.process(kafkaTransferParam, context);
                        kafkaTransferMono.subscribe(v -> {
                            // do something
                        }, t -> {
                            // do something
                            _sink.error(t);
                        });
                        break;
                    default:
                        log.error("不支持的calltype: " + model.getCallType());
                        _sink.error(new ApException("不支持的calltype: " + model.getCallType()));
                        break;
                }
                _sink.success();
            } catch (Exception e) {
                _sink.error(e);
            }
        });
    }

    /**
     * @return void
     * @Author superz
     * @Description TODO 逻辑提取到processor中
     * @Date 2021/4/21 19:16
     * @Param [model, httpCallMono]
     **/
    private void httpResp(CrossDomainModel model, Mono<ProcessResult> httpCallMono) {
        httpCallMono.subscribe(processResult -> {
            CrossDomainModel respModel = new CrossDomainModel();
            respModel.setTransId(model.getTransId());
            // 发起：HTTP_RESP
            respModel.setCallType(CallTypeEnum.HTTP_RESP);
            respModel.setPayload(BeanUtil.beanToMap(processResult));
            crossDomainChannel.message(model.getSrcZoneId(), respModel);
        }, throwable -> {
            CrossDomainModel respModel = new CrossDomainModel();
            respModel.setTransId(model.getTransId());
            respModel.setCallType(CallTypeEnum.HTTP_RESP);
            respModel.setPayload(BeanUtil.beanToMap(ProcessResult.error(throwable.getMessage())));
            crossDomainChannel.message(model.getSrcZoneId(), respModel);
        });
    }

}
